package com.boot.stream.processor;


import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;

/**
 * TopologyBuilder builder = new TopologyBuilder();
 * builder.addSource("SOURCE", "src-topic")
 * .addProcessor("PROCESS1", MyProcessor1::new , "SOURCE")
 * .addProcessor("PROCESS2", MyProcessor2::new , "PROCESS1")
 * .addProcessor("PROCESS3", MyProcessor3::new , "PROCESS1")
 * .addSink("SINK1", "sink-topic1", "PROCESS1")
 * .addSink("SINK2", "sink-topic2", "PROCESS2")
 * .addSink("SINK3", "sink-topic3", "PROCESS3");
 * 实现Processor接口并实现process，每条消息都会调用process方法
 */
@Slf4j
public class SelectKeyProcessor implements Processor<String, String> {

    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
//        this.context.schedule(Duration.ofSeconds(10), PunctuationType.WALL_CLOCK_TIME, new Punctuator() {
//            @Override
//            public void punctuate(long timestamp) {
//
//            }
//        });
    }

    /***
     * 重新定义key ，设置输出到下一级参数（可更改参数类型）
     * @param key
     * @param value
     */
    @Override
    public void process(String key, String value) {
        JSONObject obj = JSONObject.parseObject(value);
        key = obj.getString("key");
        this.context.forward(key, obj);
    }


    @Override
    public void close() {

    }
}
